package com.code.aspect;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.code.annotation.CuratorLockAnnotation;
import com.code.common.enums.ResultCode;
import com.code.common.exceptions.BusinessException;
import com.code.common.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * 分布式锁/防重复提交 aop
 *
 * @author tangyifei
 */
@Aspect
@Component
@Slf4j
public class CuratorLockAop {

    @Value("${zk.baseSleepTimeMills}")
    private int baseSleepTimeMills;

    @Value("${zk.maxRetries}")
    private int maxRetries;

    @Value("${zk.clusterAddress}")
    private String clusterAddress;

    @Value("${zk.sessionTimeoutMs}")
    private int sessionTimeoutMs;

    @Value("${zk.connectionTimeoutMs}")
    private int connectionTimeoutMs;

    /**
     * 切点，拦截被 @CuratorLockAnnotation 修饰的方法
     */
    @Pointcut("@annotation(com.code.annotation.CuratorLockAnnotation)")
    public void curatorLockPoint() {
    }

    @Around("curatorLockPoint()")
    public Object checkLock(ProceedingJoinPoint pjp) throws Throwable {
        //当前线程名
        String threadName = Thread.currentThread().getName();
        log.info("线程{}------进入分布式锁aop------", threadName);
        //获取该注解的实例对象
        CuratorLockAnnotation annotation = ((MethodSignature) pjp.getSignature()).
                getMethod().getAnnotation(CuratorLockAnnotation.class);
        //获取参数列表
        Object[] objs = pjp.getArgs();
        if (null != annotation && !ArrayUtils.isEmpty(objs)) {
            //因为只有一个JSON参数，直接取第一个
            JSONObject param = JSON.parseObject(JSON.toJSONString(objs[0]));
            //生成分布式锁key的键名，以逗号分隔
            String keyParts = annotation.keyParts();
            StringBuilder keyBuffer = new StringBuilder();
            if (StringUtil.isEmpty(keyParts)) {
                log.info("线程{} keyParts设置为空，不加锁", threadName);
                return pjp.proceed();
            } else {
                //生成分布式锁key
                String[] keyPartArray = keyParts.split(",");
                for (String keyPart : keyPartArray) {
                    keyBuffer.append(param.getString(keyPart));
                }
                String key = keyBuffer.toString();
                log.info("线程{} 要加锁的key={}", threadName, key);

                //创建zookeeper的客户端-重试策略，初始休眠时间为 1000ms, 最大重试次数为 3
                RetryPolicy retryPolicy = new ExponentialBackoffRetry(baseSleepTimeMills, maxRetries);
                // 创建一个客户端, 60000(ms)为 session 超时时间, 15000(ms)为链接超时时间
                CuratorFramework client = CuratorFrameworkFactory.newClient(clusterAddress, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
                client.start();
                //可重入排它锁-创建分布式锁, 锁空间的根节点路径为/curator/lock/**
                InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock/" + key);
                try {
                    if (mutex.acquire(-1L, null)) {
                        log.info("线程{} 获取锁成功", threadName);
                        //获得了锁, 进行业务流程
                        return pjp.proceed();
                    } else {
                        log.info("线程{} 获取锁失败", threadName);
                        throw new BusinessException(ResultCode.GET_LOCK_ERROR);
                    }
                } catch (Exception e) {
                    log.info("线程{} 获取锁失败，发生的异常：{}", threadName, e.getMessage());
                    throw new BusinessException(ResultCode.GET_LOCK_ERROR);
                } finally {
                    //完成业务流程, 释放锁
                    mutex.release();
                    log.info("线程{} 释放锁", threadName);
                    //关闭客户端
                    client.close();
                }
            }
        } else {
            return pjp.proceed();
        }
    }
}
